package com.hzccz.kafka;

public class Main {
    public static void main(String[] args) {
        
        if (args.length >= 2 && (args[0].equals("-c") || args[0].equals("--config"))) {
            ConfigUtils.loadProperties(args[1]);
            
            String[] remainingArgs = new String[args.length - 2];
            System.arraycopy(args, 2, remainingArgs, 0, remainingArgs.length);
            args = remainingArgs;
        }

        if (args.length == 0 || args[0].equals("-h") || args[0].equals("--help")) {
            printUsage();
            return;
        }

        if (args[0].equals("producer")) {
            runProducer();
        } else if (args[0].equals("consumer")) {
            runConsumer();
        } else {
            System.err.println("未知命令: " + args[0]);
            printUsage();
        }
    }

    private static void runProducer() {
        try {
            
            String bootstrapServers = ConfigUtils.getProperty("bootstrap.servers", "localhost:9092");
            String topic = ConfigUtils.getProperty("topic", "default-topic");
            String filePath = ConfigUtils.getProperty("producer.file.path", "data.txt");
            long delayMs = ConfigUtils.getLongProperty("producer.delay.ms", 1000);

            System.out.println("===== 生产者配置 =====");
            System.out.println("Kafka服务器: " + bootstrapServers);
            System.out.println("目标主题: " + topic);
            System.out.println("数据文件: " + filePath);
            System.out.println("发送延迟: " + delayMs + "ms");
            System.out.println("======================");

            KafkaFileProducer producer = new KafkaFileProducer(bootstrapServers, topic);
            producer.sendFromFile(filePath, delayMs);
            producer.close();
        } catch (Exception e) {
            System.err.println("生产者运行出错: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static void runConsumer() {
        try {
            
            String bootstrapServers = ConfigUtils.getProperty("bootstrap.servers", "localhost:9092");
            String groupId = ConfigUtils.getProperty("consumer.group.id", "default-group");
            String topic = ConfigUtils.getProperty("topic", "default-topic");
            long delayMs = ConfigUtils.getLongProperty("consumer.delay.ms", 1000);

            System.out.println("===== 消费者配置 =====");
            System.out.println("Kafka服务器: " + bootstrapServers);
            System.out.println("消费者组ID: " + groupId);
            System.out.println("目标主题: " + topic);
            System.out.println("消费延迟: " + delayMs + "ms");
            System.out.println("======================");

            KafkaFileConsumer consumer = new KafkaFileConsumer(bootstrapServers, groupId, topic);
            consumer.consumeAndPrint(delayMs);
        } catch (Exception e) {
            System.err.println("消费者运行出错: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static void printUsage() {
        System.out.println("Kafka 文件生产者和消费者工具");
        System.out.println("用法:");
        System.out.println("  生产者: java -jar kafka-file-processor.jar [-c|--config <config-file>] producer");
        System.out.println("  消费者: java -jar kafka-file-processor.jar [-c|--config <config-file>] consumer");
        System.out.println("  帮助:   java -jar kafka-file-processor.jar --help");
        System.out.println("所有配置参数都在配置文件中设置，默认使用当前目录的kafka.properties");
    }
}
